package com.intct.util;

import com.intct.common.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @author gufg
 * @since 2025-07-03 09:13
 */
public class KafkaUtil {
    /**
     * Flink Kafka输出
     * @param topic 主题名称
     * @param transactionalIdPrefix 事件ID前缀
     * @return
     */
    public static KafkaSink<String> getKafkaSink(String topic, String transactionalIdPrefix) {
        return KafkaSink.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BOOTSTRAP_SERVERS)
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic(topic)
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build())
                .setTransactionalIdPrefix(transactionalIdPrefix)
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000")
                .build();
    }

    /**
     * 从Kafka数据源获取 数据
     * @param topic
     * @param groupId
     * @return
     */
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BOOTSTRAP_SERVERS)
                .setTopics(topic)
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .build();
    }
}
